package com.deepclassifieds.daas.api.jdbc;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.sql.Types;
import java.text.MessageFormat;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.ResourceBundle;
import java.util.UUID;

import javax.xml.datatype.XMLGregorianCalendar;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.deepclassifieds.daas.api.DaasExpression;
import com.deepclassifieds.daas.api.DaasExpressionBuilder;
import com.deepclassifieds.daas.api.DaasRuntimeException;
import com.deepclassifieds.daas.api.DaasShardService;
import com.deepclassifieds.daas.api.DaasUID;
import com.deepclassifieds.daas.api.builder.ShardsBuilder;
import com.deepclassifieds.daas.api.builder.TableBuilder;
import com.deepclassifieds.daas.api.jdbc.dialect.Dialect;
import com.deepclassifieds.daas.api.jdbc.name.ColumnName;
import com.deepclassifieds.daas.api.jdbc.name.DatabaseName;
import com.deepclassifieds.daas.api.jdbc.name.TableName;
import com.deepclassifieds.daas.api.util.JAXBCommonHelper;
import com.deepclassifieds.daas.api.util.XMLGregorianCalendarHelper;
import com.deepclassifieds.daas.api.xsd.Column;
import com.deepclassifieds.daas.api.xsd.ColumnDef;
import com.deepclassifieds.daas.api.xsd.ColumnDefs;
import com.deepclassifieds.daas.api.xsd.Columns;
import com.deepclassifieds.daas.api.xsd.Database;
import com.deepclassifieds.daas.api.xsd.Databases;
import com.deepclassifieds.daas.api.xsd.ObjectFactory;
import com.deepclassifieds.daas.api.xsd.Record;
import com.deepclassifieds.daas.api.xsd.Records;
import com.deepclassifieds.daas.api.xsd.Shard;
import com.deepclassifieds.daas.api.xsd.Shards;
import com.deepclassifieds.daas.api.xsd.Table;
import com.deepclassifieds.daas.api.xsd.Tables;
import com.deepclassifieds.daas.common.Name;
import com.deepclassifieds.daas.common.Reject;
import com.deepclassifieds.daas.common.sql.ConnectionProvider;
import com.deepclassifieds.daas.common.sql.SQLExecutor;
import com.deepclassifieds.daas.common.sql.SQLTask;
import com.deepclassifieds.daas.common.util.ResourceBundleUtils;

/**
 * Will be Thread-safe.
 * @author nbarkhatov
 *
 */
public class DaasShardServiceJdbcImpl implements DaasShardService {
	
	private static final Logger logger = LoggerFactory.getLogger(DaasShardServiceJdbcImpl.class);

	private final ConnectionProvider connectionProvider;
	private final Shards shards;
	private final Dialect dialect;
	
	/** Database, Table, ColumnDef */
	private final Map<DatabaseName,Map<TableName,Map<ColumnName,ColumnDef>>> metadata; 
	//
	// DDL templates
	//
	private final String databaseCreate;
	private final String databaseDrop;
	private final String databaseUse;
	private final String tableDrop;
	private final String columnDrop;
	private final String recordSelectById;
	private final String recordSelectByUUID;
	private final String recordDeleteById;
	private final String recordDeleteByUUID;
	private final String recordSelectWhere;
	

	public DaasShardServiceJdbcImpl(Shards shards, Dialect dialect, ConnectionProvider connectionProvider) {
		Reject.ifNull(shards);
		Reject.ifNull(dialect);
		Reject.ifNull(connectionProvider);
		this.shards = shards;
		this.dialect = dialect;
		this.connectionProvider = connectionProvider;
		this.metadata = new LinkedHashMap<DatabaseName,Map<TableName,Map<ColumnName,ColumnDef>>>(); 
		ResourceBundle bundle = ResourceBundleUtils.loadResourceBundle(this.getClass(), "ddl", dialect.getLocale());
		this.databaseCreate     = bundle.getString("database.create");
		this.databaseDrop       = bundle.getString("database.drop");
		this.databaseUse        = bundle.getString("database.use");
		this.tableDrop          = bundle.getString("table.drop");
		this.columnDrop         = bundle.getString("column.drop");
		this.recordSelectById   = bundle.getString("record.select.by.id");
		this.recordSelectByUUID = bundle.getString("record.select.by.uuid");
		this.recordDeleteById   = bundle.getString("record.delete.by.id");
		this.recordDeleteByUUID = bundle.getString("record.delete.by.uuid");
		this.recordSelectWhere  = bundle.getString("record.select.where");
	}
	
	// ----------------------------------------------------------------------------------------------------------------
	//   interface implementation methods below
	// ----------------------------------------------------------------------------------------------------------------

	@Override
	public Databases getDatabases() {
		throw new DaasRuntimeException("not implemented yet...");
	}

	@Override
	public Database getDatabaseByName(Name databaseName) {
		throw new DaasRuntimeException("not implemented yet...");
	}

	@Override
	public Database createDatabase(final Name databaseName, final Shard shard) {
		Reject.ifNull(databaseName);
		return new SQLExecutor<Database>(connectionProvider).execute(new SQLTask<Database>() {
			@Override
			public Database execute(Connection conn) throws SQLException {
		    	Statement stmt = conn.createStatement();
		    	DatabaseName databaseNameInternal = new DatabaseName(databaseName, shard);
		    	stmt.execute(debug(MessageFormat.format(databaseCreate, databaseNameInternal)));
		    	stmt.close();
		    	Database database = new Database();
		    	database.setName(databaseName.toString());
		    	database.setTables(new Tables());
		    	database.setShards(shards);
		    	//
		    	// update metadata
		    	//
		    	metadata.put(databaseNameInternal, new LinkedHashMap<TableName,Map<ColumnName,ColumnDef>>());
		    	return database;
			}
		});
	}

	@Override
	public void dropDatabase(final Database database, final Shard shard) {
		Reject.ifNull(database);
		new SQLExecutor<Void>(connectionProvider).execute(new SQLTask<Void>() {
			@Override
			public Void execute(Connection conn) throws SQLException {
		    	Statement stmt = conn.createStatement();
		    	DatabaseName databaseName = new DatabaseName(new Name(database.getName()), shard);
		    	stmt.execute(debug(MessageFormat.format(databaseDrop, databaseName)));
		    	stmt.close();
		    	//
		    	// update metadata
		    	//
		    	metadata.remove(databaseName);
		    	return null;
			}
		});
	}

	@Override
	public Shards getShards() {
		throw new DaasRuntimeException("not applicable...");
	}

	@Override
	public Shard getShardByNumber(int shardNumber) {
		throw new DaasRuntimeException("not applicable...");
	}

	@Override
	public Tables getTables(Database database) {
		throw new DaasRuntimeException("not implemented yet...");
	}

	@Override
	public Table getTableByName(Database database, Shard shard, Name tableName) {
		Map<TableName,Map<ColumnName,ColumnDef>> tableMap = metadata.get(new DatabaseName(new Name(database.getName()), shard));
		if(tableMap == null) throw new DaasRuntimeException("No such database '" + database.getName() + "'");
		Map<ColumnName,ColumnDef> columnMap = tableMap.get(new TableName(tableName, shard));
		if(columnMap == null) throw new DaasRuntimeException("No such table '" + tableName + "' in database '" + database.getName() + "'");
		TableBuilder tableBuilder = new TableBuilder(new Name(database.getName()), tableName);
		for(Map.Entry<ColumnName,ColumnDef> entry : columnMap.entrySet()) {
			tableBuilder.addColumnDef(entry.getValue());
		}
		return tableBuilder.build();
	}

//	@Override
//	public Table createTable(final Database database, final Name tableName, final ColumnDefs columnDefs) {
//		throw new DaasRuntimeException("cannot call this method...");
//	}
	
	@Override
	public Table createTable(final Database database, final Shard shard, final Name tableName, final ColumnDefs columnDefs) {
		Reject.ifNull(database);
		Reject.ifNull(tableName);
		Reject.ifNull(columnDefs);
		Reject.ifNull(columnDefs.getColumnDef());
		return new SQLExecutor<Table>(connectionProvider).execute(new SQLTask<Table>() {
			@Override
			public Table execute(Connection conn) throws SQLException {
		    	Statement stmt = conn.createStatement();
		    	DatabaseName theDatabaseName = new DatabaseName(new Name(database.getName()), shard);
		    	TableName theTableName = new TableName(tableName, shard);
		    	stmt.addBatch(debug(database, MessageFormat.format(databaseUse, theDatabaseName)));
		    	stmt.addBatch(debug(database, dialect.createTable(theTableName, columnDefs)));
		    	stmt.executeBatch();
		    	stmt.close();
		    	Table table = new Table();
		    	table.setDatabase(database.getName());
		    	table.setColumnDefs(columnDefs);
		    	table.setName(tableName.toString());
		    	//
		    	// update metadata
		    	//
		    	Map<ColumnName,ColumnDef> columnDefMap = new LinkedHashMap<ColumnName,ColumnDef>(columnDefs.getColumnDef().size());
		    	for(ColumnDef columnDef : columnDefs.getColumnDef()) {
		    		columnDefMap.put(new ColumnName(new Name(columnDef.getName())), columnDef);
		    	}
		    	getTableMap(theDatabaseName).put(theTableName, columnDefMap);
		    	return table;
			}
		});
	}
	
//	@Override
//	public void alterTableAddColumn(final Database database, final Table table, final ColumnDef columnDef) {
//		throw new DaasRuntimeException("cannot call this method...");
//	}

	@Override
	public void alterTableAddColumn(final Database database, final Shard shard, final Table table, final ColumnDef columnDef) {
		Reject.ifNull(database);
		Reject.ifNull(table);
		Reject.ifNull(columnDef);
		new SQLExecutor<Void>(connectionProvider).execute(new SQLTask<Void>() {
			@Override
			public Void execute(Connection conn) throws SQLException {
		    	Statement stmt = conn.createStatement();
		    	DatabaseName  theDatabaseName = new DatabaseName(new Name(database.getName()), shard);
		    	TableName theTableName = new TableName(new Name(table.getName()), shard);
		    	stmt.addBatch(debug(database, MessageFormat.format(databaseUse, theDatabaseName)));
		    	stmt.addBatch(debug(database, dialect.alterTableAddColumn(theTableName, columnDef)));
		    	stmt.executeBatch();
		    	stmt.close();
		    	//
		    	// update cached metadata
		    	//
		    	getColumnDefMap(theDatabaseName, theTableName).put(new ColumnName(new Name(columnDef.getName())), columnDef);
		    	return null;
			}
		});
	}
	

//	@Override
//	public void alterTableDropColumn(final Database database, final Table table, final Name columnName) {
//		throw new DaasRuntimeException("cannot call this method...");
//	}

	@Override
	public void alterTableDropColumn(final Database database, final Shard shard, final Table table, final Name columnName) {
		Reject.ifNull(database);
		Reject.ifNull(table);
		Reject.ifNull(columnName);
		new SQLExecutor<Void>(connectionProvider).execute(new SQLTask<Void>() {
			@Override
			public Void execute(Connection conn) throws SQLException {
		    	Statement stmt = conn.createStatement();
		    	DatabaseName theDatabaseName = new DatabaseName(new Name(database.getName()), shard);  
		    	TableName theTableName = new TableName(new Name(table.getName()), shard);
		    	stmt.addBatch(debug(database, MessageFormat.format(databaseUse, theDatabaseName)));
		    	stmt.addBatch(debug(database, MessageFormat.format(columnDrop, theTableName, columnName.toString())));
		    	stmt.executeBatch();
		    	stmt.close();
		    	//
		    	// update cached metadata
		    	//
		    	getColumnDefMap(theDatabaseName, theTableName).remove(columnName);
		    	return null;
			}
		});
	}	
	
//	@Override
//	public void dropTable(final Database database, final Table table) {
//		throw new DaasRuntimeException("cannot call this method...");
//	}
	
	@Override
	public void dropTable(final Database database, final Shard shard, final Table table) {
		Reject.ifNull(database);
		Reject.ifNull(table);
		new SQLExecutor<Void>(connectionProvider).execute(new SQLTask<Void>() {
			@Override
			public Void execute(Connection conn) throws SQLException {
		    	DatabaseName theDatabaseName = new DatabaseName(new Name(database.getName()), shard);  
				TableName theTableName = new TableName(new Name(table.getName()), shard);
		    	Statement stmt = conn.createStatement();
		    	stmt.addBatch(debug(database, MessageFormat.format(databaseUse, theDatabaseName)));
		    	stmt.addBatch(debug(database, MessageFormat.format(tableDrop, theTableName)));
		    	stmt.executeBatch();
		    	stmt.close();
		    	//
		    	// update cached metadata
		    	//
		    	getTableMap(new DatabaseName(new Name(database.getName()), shard)).remove(theTableName);
		    	return null;
			}
		});
	}
	
	//
	// DML
	//

	@Override
	public Record insertRecord(final Database database, final Shard shard, final Table table, final Record record) {
		Reject.ifNull(database);
		Reject.ifNull(shard);
		Reject.ifNull(table);
		Reject.ifNull(record);
		Reject.ifNull(record.getColumns());
		return new SQLExecutor<Record>(connectionProvider).execute(new SQLTask<Record>() {
			@Override
			public Record execute(Connection conn) throws SQLException {
				//
				// TODO: blend JDBC metadata into record here
				//
				Record inserted = JAXBCommonHelper.copy(new ObjectFactory().createRecord(record), Record.class);
				Column identityColumn = new Column();
				
		    	Statement stmt = conn.createStatement();
		    	DatabaseName theDatabaseName = new DatabaseName(new Name(database.getName()), shard);
		    	stmt.execute(debug(database, table, shard, MessageFormat.format(databaseUse, theDatabaseName)));
		    	stmt.close();
		    	TableName theTableName = new TableName(new Name(table.getName()), shard);
		    	String insertRecordDML = debug(database, table, shard, insertRecordDML(theTableName, record.getColumns()));
		    	PreparedStatement pstmt = identityColumn != null ?
			    		conn.prepareStatement(insertRecordDML, Statement.RETURN_GENERATED_KEYS) :
				    		conn.prepareStatement(insertRecordDML);
		    	//
		    	// map columns into prepared statement
		    	//
		    	int parameterIndex = 1;
				for(Column column : record.getColumns().getColumn()) {
					if(column != identityColumn) {						
						String value = column.getValue();
						ColumnDef columnDef = getColumnDef(
								theDatabaseName, 
								theTableName, 
								new ColumnName(new Name(column.getName())));
						switch(columnDef.getDatatype()) {
						case BOOLEAN:  
							if(value == null) {
								pstmt.setNull(parameterIndex++,        Types.BOOLEAN);
							} else {
								pstmt.setBoolean(parameterIndex++,     Boolean.parseBoolean(value)); 
							}
							break;
						case BYTE:
							if(value == null) {
								pstmt.setNull(parameterIndex++,        Types.TINYINT);
							} else {
								pstmt.setByte(parameterIndex++,        Byte.parseByte(value)); 
							}
							break;
						case SHORT:
							if(value == null) {
								pstmt.setNull(parameterIndex++,        Types.SMALLINT);
							} else {
								pstmt.setShort(parameterIndex++,       Short.parseShort(value)); 
							}
							break;
						case INTEGER:
							if(value == null) {
								pstmt.setNull(parameterIndex++,        Types.INTEGER);
							} else {
								pstmt.setShort(parameterIndex++,       Short.parseShort(value)); 
							}
							break;
						case LONG:
							if(value == null) {
								pstmt.setNull(parameterIndex++,        Types.BIGINT);
							} else {
								pstmt.setLong(parameterIndex++,        Long.parseLong(value)); 
							}
							break;
						case FLOAT:
							if(value == null) {
								pstmt.setNull(parameterIndex++,        Types.FLOAT);
							} else {
								pstmt.setFloat(parameterIndex++,       Float.parseFloat(value)); 
							}
							break;
						case DOUBLE:
							if(value == null) {
								pstmt.setNull(parameterIndex++,        Types.DOUBLE);
							} else {
								pstmt.setDouble(parameterIndex++,      Double.parseDouble(value)); 
							}
							break;
						case DECIMAL:
							if(value == null) {
								pstmt.setNull(parameterIndex++,        Types.DECIMAL);
							} else {
								pstmt.setBigDecimal(parameterIndex++,  new BigDecimal(value)); 
							}
							break;
						case STRING:   
							if(value == null) {
								pstmt.setNull(parameterIndex++,        Types.VARCHAR);
							} else {
								pstmt.setString(parameterIndex++,      value); 
							}
							break;
						case DATETIME:
							if(value == null) {
								pstmt.setNull(parameterIndex++,        Types.TIMESTAMP);
							} else {
								pstmt.setTimestamp(parameterIndex++,   new Timestamp(XMLGregorianCalendarHelper.toDate(XMLGregorianCalendarHelper.parse(value)).getTime())); 
							}
							break;
						}
					}
				}
		    	//
		    	// map system columns
		    	//
		    	String uuid = UUID.randomUUID().toString();
		    	Date dateNow = new Date();
		    	Timestamp timestampNow = new Timestamp(dateNow.getTime());
		    	XMLGregorianCalendar xcalNow = XMLGregorianCalendarHelper.toXMLGregorianCalendar(dateNow);
		    	int version = 0;
		    	inserted.setUuid(uuid);
		    	inserted.setCreationDate(xcalNow);
		    	inserted.setLastModificationDate(xcalNow);
		    	inserted.setVersion(version);
		    	pstmt.setString(parameterIndex++, uuid); // DAAS_UUID
		    	pstmt.setTimestamp(parameterIndex++, timestampNow); // DAAS_CREATION_DATE
		    	pstmt.setTimestamp(parameterIndex++, timestampNow); // DAAS_LAST_MODIFICATION_DATE
		    	pstmt.setInt(parameterIndex++, version); // DAAS_VERSION
		    	//
		    	// insert
		    	//
		    	pstmt.executeUpdate();
		    	//
		    	// populate identity column if any
		    	//
		    	if(identityColumn != null) {
			    	ResultSet rs = pstmt.getGeneratedKeys();
			    	if(rs.next()) {
				    	BigDecimal id = rs.getBigDecimal(1); // get automatically generated key
//				    	long globalId = IdentityHelper.toGlobalId(shard, id.longValue());
//				    	identityColumn.setValue(Long.toString(globalId));
				    	long uid = new DaasUID(shard.getNumber(), id.longValue()).getValue();
				    	inserted.setUid(uid);
			    	}
			    	rs.close();                          
		    	}
		    	pstmt.close(); 
		    	return inserted;
			}
		});
	}

	@Override
	public Record selectRecordByUID(final Database database, final Table table, final DaasUID daasUID) {
		Reject.ifNull(database);
		Reject.ifNull(table);
		Reject.ifNull(daasUID);
		return new SQLExecutor<Record>(connectionProvider).execute(new SQLTask<Record>() {
			@Override
			public Record execute(Connection conn) throws SQLException {				
		    	Statement stmt = conn.createStatement();
		    	//
		    	// use database
		    	//
		    	Shard shard = new Shard();
		    	shard.setNumber(daasUID.getShardNumber());
		    	DatabaseName theDatabaseName = new DatabaseName(new Name(database.getName()), shard);  
		    	stmt.execute(debug(database, MessageFormat.format(databaseUse, theDatabaseName)));
		    	//
		    	// select by id
		    	//
		    	TableName theTableName = new TableName(new Name(table.getName()), shard);
		    	String idColumnName = dialect.resolveIdentityColumnName(theTableName, conn);
		    	//String idValue = Long.toString(IdentityHelper.toLocalId(uid));
		    	String idValue = Long.toString(daasUID.getId());
		    	ResultSet rs = stmt.executeQuery(debug(database, MessageFormat.format(recordSelectById, theTableName, idColumnName, idValue)));
	    		Record record = null;
		    	if(rs.next()) {
		    		record = dialect.toRecord(rs);
		    	}
		    	stmt.close();
		    	return record;
			}
		});
	}	

	@Override
	public Record selectRecordByUUID(final Database database, final Shard shard, final Table table, final UUID uuid) {
		Reject.ifNull(database);
		Reject.ifNull(shards); // not used
		Reject.ifNull(table);
		Reject.ifNull(uuid);
		return new SQLExecutor<Record>(connectionProvider).execute(new SQLTask<Record>() {
			@Override
			public Record execute(Connection conn) throws SQLException {
		    	Statement stmt = conn.createStatement();
		    	DatabaseName theDatabaseName = new DatabaseName(new Name(database.getName()), shard);
		    	stmt.execute(debug(database, MessageFormat.format(databaseUse, theDatabaseName)));
		    	TableName theTableName = new TableName(new Name(table.getName()), shard);
		    	ResultSet rs = stmt.executeQuery(debug(database, MessageFormat.format(recordSelectByUUID, theTableName, uuid.toString())));
	    		Record record = null;
		    	if(rs.next()) {
		    		record = dialect.toRecord(rs);
		    	}
		    	stmt.close();
		    	return record;
			}
		});
	}	
	
	@Override
	public DaasExpressionBuilder createExpressionBuilder() {
		return new DaasExpressionBuilderJdbcImpl();
	}

	@Override
	public Records selectRecords(final Database database, final Shard shard, final Table table, final DaasExpression where) {
		Reject.ifNull(database);
		Reject.ifNull(shards); // not used
		Reject.ifNull(table);
		Reject.ifNull(where);
		return new SQLExecutor<Records>(connectionProvider).execute(new SQLTask<Records>() {
			@Override
			public Records execute(Connection conn) throws SQLException {
		    	Statement stmt = conn.createStatement();
		    	DatabaseName theDatabaseName = new DatabaseName(new Name(database.getName()), shard);
		    	stmt.execute(debug(database, MessageFormat.format(databaseUse, theDatabaseName)));
		    	TableName theTableName = new TableName(new Name(table.getName()), shard);
		    	ResultSet rs = stmt.executeQuery(debug(database, MessageFormat.format(recordSelectWhere, theTableName, where.toString())));
	    		Records records = new Records();
		    	while(rs.next()) {
		    		records.getRecord().add(dialect.toRecord(rs));
		    	}
		    	stmt.close();
		    	return records;
			}
		});
	}

	@Override
	public void updateRecord(Database database, Shard shard, Table table, Record record) {
		throw new DaasRuntimeException("not implemented yet...");
	}

	@Override
	public void deleteRecordByUID(final Database database, final Table table, final DaasUID daasUID) {
		Reject.ifNull(database);
		Reject.ifNull(table);
		new SQLExecutor<Void>(connectionProvider).execute(new SQLTask<Void>() {
			@Override
			public Void execute(Connection conn) throws SQLException {
		    	Statement stmt = conn.createStatement();
		    	//
		    	// use database
		    	//
		    	Shard shard = new ShardsBuilder().createAndAddShard(daasUID.getShardNumber());
		    	DatabaseName theDatabaseName = new DatabaseName(new Name(database.getName()), shard);  
		    	stmt.execute(debug(database, MessageFormat.format(databaseUse, theDatabaseName)));
		    	//
		    	// select by id
		    	//
		    	TableName theTableName = new TableName(new Name(table.getName()), shard); 
		    	String idColumnName = dialect.resolveIdentityColumnName(theTableName, conn);
		    	String idValue = Long.toString(daasUID.getId()); //IdentityHelper.toLocalId(id));
		    	stmt.executeUpdate(debug(database, MessageFormat.format(recordDeleteById, theTableName, idColumnName, idValue)));
		    	stmt.close();
		    	return null;
			}
		});
	}

	@Override
	public void deleteRecordByUUID(final Database database, final Shard shard, final Table table, final UUID uuid) {
		Reject.ifNull(database);
		Reject.ifNull(table);
		Reject.ifNull(uuid);
		new SQLExecutor<Void>(connectionProvider).execute(new SQLTask<Void>() {
			@Override
			public Void execute(Connection conn) throws SQLException {
		    	Statement stmt = conn.createStatement();
		    	DatabaseName theDatabaseName = new DatabaseName(new Name(database.getName()), shard);
		    	TableName theTableName = new TableName(new Name(table.getName()), shard);
		    	stmt.addBatch(debug(database, MessageFormat.format(databaseUse, theDatabaseName)));
		    	stmt.addBatch(debug(database, MessageFormat.format(recordDeleteByUUID, theTableName, uuid.toString())));
		    	stmt.executeBatch();
		    	stmt.close();
		    	return null;
			}
		});
	}

	// ----------------------------------------------------------------------------------------------------------------
	//   helper methods below
	// ----------------------------------------------------------------------------------------------------------------

	private String insertRecordDML(TableName tableName, Columns columns) {
		Reject.ifNull(tableName);
		Reject.ifNull(columns);
		int size = columns.getColumn().size();
		StringBuilder sb = new StringBuilder(64 * size + 256);
		sb.append("INSERT INTO ").append(tableName).append(" (\n");
		int columnCount = 0;
		for (Column column : columns.getColumn()) {
//			if(!(column.getDatatype() == Datatype.AUTO_GENERATED_IDENTITY)) {
				sb.append("  ").append(column.getName()).append(",\n");	
				++columnCount;
//			}
		}
		//sb.append("  ").append(ColumnName.DAAS_UID).append(",\n");
		sb.append("  ").append(Dialect.DaasColumnName.DAAS_UUID).append(",\n");
		sb.append("  ").append(Dialect.DaasColumnName.DAAS_CREATION_DATE).append(",\n");
		sb.append("  ").append(Dialect.DaasColumnName.DAAS_LAST_MODIFICATION_DATE).append(",\n");
		sb.append("  ").append(Dialect.DaasColumnName.DAAS_VERSION).append("\n");
		sb.append(") VALUES (");
		for (int i = 0; i < columnCount ; ++i) {
			sb.append("?, ");			
		}
		sb.append("?, ?, ?, ?");
		sb.append(")");
		return sb.toString();
	}

	private String debug(String ddl) {
		Reject.ifNull(ddl);
		logger.debug("{}:// - {}", dialect.getLocale(), ddl);
		return ddl;
	}

	private String debug(Database database, String ddl) {
		Reject.ifNull(database);
		Reject.ifNull(ddl);
		logger.debug("{}://{} - {}", new Object[] {
				dialect.getLocale(), database.getName(), ddl});
		return ddl;
	}

	private String debug(Database database, Table table, Shard shard, String ddl) {
		Reject.ifNull(database);
		Reject.ifNull(table);
		Reject.ifNull(ddl);
		logger.debug("{}://{}/{}/{} - {}", new Object[] {
				dialect.getLocale(), database.getName(), shard.getNumber(), table.getName(), ddl});
		return ddl;
	}
	
	private Map<TableName,Map<ColumnName,ColumnDef>> getTableMap(DatabaseName databaseName) {
		assert databaseName != null;
    	Map<TableName,Map<ColumnName,ColumnDef>> tableMap = metadata.get(databaseName);
    	if (tableMap == null) {
    		throw new DaasRuntimeException("Metadata for database '" + databaseName + "' not found in local cache");
    	}
    	return tableMap;
	}

	private Map<ColumnName,ColumnDef> getColumnDefMap(DatabaseName databaseName, TableName tableName) {
		assert databaseName != null;
		assert tableName != null;
    	Map<TableName,Map<ColumnName,ColumnDef>> tableMap = getTableMap(databaseName);
    	Map<ColumnName,ColumnDef> columnDefMap = tableMap.get(tableName);
    	if (columnDefMap == null) {
    		throw new DaasRuntimeException("Metadata for table '" + tableName + 
    				"' in database '" + databaseName + "' not found in local cache");
    	}
    	return columnDefMap;
	}

	private ColumnDef getColumnDef(DatabaseName databaseName, TableName tableName, ColumnName columnName) {
		assert databaseName != null;
		assert tableName != null;
		assert columnName != null;
		Map<ColumnName,ColumnDef> columnDefMap = getColumnDefMap(databaseName, tableName);
		ColumnDef columnDef = columnDefMap.get(columnName);
    	if (columnDef == null) {
    		throw new DaasRuntimeException("Metadata for column '" + columnName + 
    				"' in table '" + tableName + 
    				"' in database '" + databaseName + "' not found in local cache");
    	}
    	return columnDef;
	}

}

